package com.shujia.flink.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

object Demo2Mysql {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val ds: DataStream[String] = env.socketTextStream("master", 7777)


    //使用自定义sink
    ds.addSink(new MysqlSinkFunction)


    env.execute()

  }

}

class MysqlSinkFunction extends RichSinkFunction[String] {

  /**
    * open方法在invoke之前执行，而且每一个分区只执行一次
    * 代码一启动就会调用
    *
    */

  var con: Connection = _

  override def open(parameters: Configuration): Unit = {
    println("创建连接")
    //1、加载驱动
    Class.forName("com.mysql.jdbc.Driver")

    //2\ 创建连接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/student?useUnicode=true&characterEncoding=UTF-8", "root", "123456")

  }


  /**
    * invoke:每一条数据都会执行一次
    * 如果将连接放在invoke,每一条数据都会创建一个连接
    *
    */
  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {


    //3 插入数据
    val stat: PreparedStatement = con.prepareStatement("insert into student (id,name,age,gender,clazz) values(?,?,?,?,?)")

    val split: Array[String] = value.split(",")


    stat.setInt(1, split(0).toInt)
    stat.setString(2, split(1))
    stat.setInt(3, split(2).toInt)
    stat.setString(4, split(3))
    stat.setString(5, split(4))


    stat.execute()


  }

  /**
    * close方法在invoke之后执行，而且每一个分区只执行一次
    *
    * 程序关闭的时候调用
    *
    */

  override def close(): Unit = {
    con.close()
  }
}

